package com.mq.consumer.test;

import com.mq.common.MqConnectionUtils;
import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * @author zhanghaidong
 * @version v1.0
 * @description
 * @date 2019/8/16
 */
public class Consumer {

    public static void main(String[] args) {

        try {
            Connection connection = MqConnectionUtils.getConnection();

            Channel channel = connection.createChannel();
            System.out.println("信道id = " + channel.getChannelNumber());

            String exchangeName = "direct-exchange";
            channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT);

            /**
             * exclusive：设为true，队列将变成私有的，此时只有你的应用程序才能消费队列消息
             *             【当你想要限制一个队列只有一个消费者的时候很有帮助】
             * autoDelete：当最后一个消费者取消订阅的时候，队列就会自动删除
             *
             *      如果你需要临时队列只为一个消费者服务，结合使用autoDelete和exclusive，当消费者断开连接时，队列就被移除了
             */
            channel.queueDeclare("direct-queue", false, false, false, null);

            /**
             * 只是检查队列是否存在，如果队列存在则正常返回；否则会抛出一个异常，并且执行该操作对Channel不再
             * 可用，后续应该创建新的Channel对象使用。
             */
//            AMQP.Queue.DeclareOk declareOk = channel.queueDeclarePassive("direct-queue");
//            System.out.println(declareOk.getMessageCount());
//            System.out.println(declareOk.getConsumerCount());

            /**
             * queuePurge不删除队列，而是清空队列中数据
             */
            //channel.queuePurge("direct-queue");

            // 交换器和队列绑定
            channel.queueBind("direct-queue", "direct-exchange", "direct-routingKey", null);


            //========================================================================================================
            //
            // 在RabbitMQ中，消费者可以以推模式和拉模式两种方式消费消息，分别对象basicConsume方法和basicGet方法。
            //
            //         推模式中，由服务端主动推送消息到消费这，消费者在会调方法中处理消息，然后响应服务端；
            //         拉模式中，消费者每次主动从队列中拉起单条消息消费，然后响应服务端。
            //
            //========================================================================================================
            //
            //   1、通过AMQP的basic.consume命令订阅，这样会将信道置为接收模式，直到取消对队列的订阅。订阅了消息后，消费者在消费（或者拒绝）
            //      最近接收的那条消息后，就能从队列（可用的）中自动接收下一条消息
            //   2、只想获取单条消息，basic.get(订阅消息---获取消息---取消订阅)
            //
            //   3、订阅队列的时候就将auto_ack参数设置为true
            //   4、通过basic.ack显示确认
            //
            //   5、如果消费者收到一条消息，然后确认之前从rabbit断开或者取消订阅，RabbitMQ会认为这条消息没有分发，然后重新分发给下一个订阅者
            //      这样确保当你的程序奔溃，消息会被下发给另一个消费者处理
            //   6、应用程序忘记确认，rabbitMQ将不会给该消费者发送更多消息。这是因为在上一条消息被确认前，rabbit会认为这个消费者没有准备好接收下一条
            //
            //
            //=============================================推模式=====================================================

            /**
             * queue:   指定队列名称，即消费哪个队列里的数据。
             * autoAck: 是否自动进行消息确认。设置为true表示自动消息确认，消费者收到消息时，自动向服务端发送ack信号，服务端收到该信号，则认为消息已经被消费，将其从队列中删除。
             *          设置为false值，需要消费者手动进行ack。
             * consumerTag: 消费者标签，用了区分多个消费者。
             * noLocal: 设置为true，表示不能将同一个Connection中生成者发送的消息传递给这个Connection中的消费者。
             * exclusive: 是否排他。
             * arguments: 其他结构化参数。
             * callback: 消费者会调函数，处理发送过来的消息。
             * cancelCallback: 消费者取消订阅时的回调方法。
             * shutdownSignalCallback: channel或者connection关闭时的回调方法。
             */
            //consumerMethod1(channel);

            // DefaultConsumer 实现了 Consumer
            channel.basicConsume("direct-queue", new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String key = envelope.getRoutingKey();
                    String message = new String(body);
                    long deliveryTag = envelope.getDeliveryTag();

                    System.out.println("routingKey = " + key + " ; message = " + message + " ; deliveryTag = " + deliveryTag);

                    // only consumer io warning messages
                    if (message.startsWith("Hello")) {
                        //consume message
                        channel.basicAck(deliveryTag,false);

                    } else { //reject other messages and requeue them
                        /**
                         * requeue: 设置为true，rabbit会把消息重新发送给下一个消费者
                         *                false rabbit会把消息从队列移除
                         *
                         *  死信队列 reject + requeue=false
                         */
                        channel.basicReject(deliveryTag,true);
                    }
                }
            });



            // ==============================拉模式============================================================
            /**
             * deliveryTag: 消息的编号
             * requeue: 消费者拒绝后，服务端是否将消息重新入队
             * multiple:
             *      basicNack，multiple参数设置为true时，表明一次性拒绝多条消息：deliveryTag编号之前的所有未被消费的消息都被拒绝。
             */

            //GetResponse response = channel.basicGet("direct-queue", true);

            //GetResponse getResponse = channel.basicGet("direct-queue", false);
            //消费
            //handleMessage(getResponse);
            //channel.basicAck(getResponse.getEnvelope().getDeliveryTag(), false);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static void consumerMethod1(Channel channel) throws IOException {
        String result = channel.basicConsume("direct-queue", true, "", false, false, null,
                // deliverCallback callback when a message is delivered
                (consumerTag, message) -> {
                    System.out.println(consumerTag + "----" + new String(message.getBody()));
                },
                // callback when the consumer is cancelled
                consumerTag -> {

                },
                // callback when the channel/connection is shut down
                (consumerTag, sig) -> {

                });
    }

    private static void handleMessage (GetResponse response) {

    }
}
